package com.winit.core.websocket.handler;

import com.alibaba.fastjson.JSONObject;
import com.winit.common.util.SpringContextHolder;
import com.winit.core.mq.broker.MessageBroker;
import com.winit.core.mq.listener.QueueListener;
import com.winit.core.mq.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;
import org.springframework.web.socket.*;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by xiangyu.liang on 2016/8/17.
 */

public class MessageQueueHandler implements WebSocketHandler {

    protected MessageBroker messageBroker = SpringContextHolder.getBean(MessageBroker.class);
    private Map<WebSocketSession, List<QueueListener>> sessionToListenerMap = new ConcurrentHashMap<>();
    private static final Logger logger = LoggerFactory.getLogger(MessageQueueHandler.class);

    @Override
    public void afterConnectionEstablished(WebSocketSession webSocketSession) throws Exception {
        logger.info("ws连接成功");
        this.sessionToListenerMap.put(webSocketSession, new ArrayList<QueueListener>());
        logger.info("ws目前保持了:" + sessionToListenerMap.keySet().size() + " 个session链接.");
    }

    @Override
    public void handleMessage(final WebSocketSession webSocketSession, final WebSocketMessage<?> webSocketMessage) throws Exception {
        String message = (String) webSocketMessage.getPayload();
        if (StringUtils.isEmpty(message)) {
            return;
        }
        final String queueName = message;
        // 基于消息队列的Websocket监听器,用于监听消息队列, 并试图向ws的session推送消息.
        logger.info("创建ws消息监听,queueName=" + queueName);
        registerListener(webSocketSession, queueName);
    }

    protected void registerListener(final WebSocketSession webSocketSession, final String queueName) {
        //创建监听
        QueueListener listener = new QueueListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    logger.info("队列[" + queueName + "]消息发送开始." + JSONObject.toJSONString(message));
                    WebSocketMessage wsMessage = new TextMessage(JSONObject.toJSONString(message));
                    webSocketSession.sendMessage(wsMessage);
                    logger.info("队列[" + queueName + "]消息发送结束." + JSONObject.toJSONString(message));
                } catch (Exception e) {
                    logger.error("队列[" + queueName + "]消息发送失败." + JSONObject.toJSONString(message), e);
                    //logger.error(e.toString());
                }
            }
        };
        //绑定session和listener的关系
        List<QueueListener> listeners = sessionToListenerMap.get(webSocketSession);
        if (listeners == null) {
            listeners = new ArrayList<>();
            sessionToListenerMap.put(webSocketSession, listeners);
        }
        listeners.add(listener);
        //绑定queue和listener的关系
        messageBroker.addListener(queueName, listener, true);
    }

    @Override
    public void handleTransportError(WebSocketSession webSocketSession, Throwable throwable) throws Exception {
        logger.info("ws异常退出");
        if (webSocketSession.isOpen()) {
            webSocketSession.close();
        }
        removeListenerBySession(webSocketSession);
        logger.info("ws目前保持了:" + sessionToListenerMap.keySet().size() + " 个session链接.");
    }

    @Override
    public void afterConnectionClosed(WebSocketSession webSocketSession, CloseStatus closeStatus) throws Exception {
        logger.info("ws连接关闭");
        this.removeListenerBySession(webSocketSession);
        this.sessionToListenerMap.remove(webSocketSession);
        logger.info("ws目前保持了:" + sessionToListenerMap.keySet().size() + " 个session链接.");
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    private void removeListenerBySession(WebSocketSession session) {
        List<QueueListener> listeners = sessionToListenerMap.get(session);
        if (null == listeners) {
            return;
        }
        messageBroker.removeListeners(listeners);
    }
}
